package com.bfxy.disruptor.common;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

//都是对于消费者线程的
//有锁的方式
public final class BlockingWaitStrategy implements WaitStrategy {
    private final Lock lock = new ReentrantLock();//重入锁
    //Condition用于多线程的唤醒和阻塞。类似于java的wait和notify，
    private final Condition processorNotifyCondition = lock.newCondition();

    //EventProcessor来消费数据，单消费者的实现是BatchEventProcessor<T>，waitFor方法是在BatchEventProcessor里面调用的。
    /*在BatchEventProcessor的run方法，
    final long availableSequence = sequenceBarrier.waitFor(nextSequence);*/
    
    @Override
    public long waitFor(long sequence, Sequence cursorSequence, Sequence dependentSequence, 
    		SequenceBarrier barrier) throws AlertException, InterruptedException {
        long availableSequence;
        //sequence是下一个消费者想要的序号，cursorSequence是生产者序号（RingBuffer里面最大的序号值），小于 消费者想要的序号sequence
        if ((availableSequence = cursorSequence.get()) < sequence) {
            lock.lock();
            try {   //是否有可用的Sequence，没有可用的，Condition就一直等着，
                while ((availableSequence = cursorSequence.get()) < sequence) {
                    barrier.checkAlert();
                    processorNotifyCondition.await();//等着
                }
            }
            finally {
                lock.unlock();
            }
        }
        
        //唤醒之后，就解锁，就去消费。
        while ((availableSequence = dependentSequence.get()) < sequence) {
            barrier.checkAlert();
        }

        return availableSequence;
    }

    @Override
    public void signalAllWhenBlocking() {
        lock.lock();
        try {   //当前有可用的序号，就唤醒阻塞的线程
            processorNotifyCondition.signalAll();
        }
        finally {
            lock.unlock();
        }
    }
}
